package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import javax.xml.crypto.Data;

public class Demo3Filter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> studentsDS = env.readTextFile("flink/data/students.txt");

        //取出性别为男的学生
        DataStream<String> filterDS = studentsDS
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        String sex = value.split(",")[3];
                        return "男".equals(sex);
                    }
                });


        //filterDS.print();

        DataStream<String> filterDS2 = studentsDS
                .filter(stu -> "女".equals(stu.split(",")[3]));

        filterDS2.print();

        env.execute();
    }
}
